package sindice.util;

import java.io.InputStream;
import java.io.StringWriter;
import java.io.UnsupportedEncodingException;
import java.net.URL;
import java.net.URLEncoder;
import java.text.NumberFormat;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.atomic.AtomicBoolean;

import org.apache.commons.lang.StringUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import com.hp.hpl.jena.rdf.arp.ARP;
import com.hp.hpl.jena.rdf.arp.JenaReader;
import com.hp.hpl.jena.rdf.model.Model;
import com.hp.hpl.jena.rdf.model.ModelFactory;
import com.hp.hpl.jena.rdf.model.RDFReader;
import com.hp.hpl.jena.rdf.model.ResIterator;
import com.hp.hpl.jena.rdf.model.Resource;
import com.hp.hpl.jena.rdf.model.Statement;
import com.hp.hpl.jena.rdf.model.StmtIterator;

public class SindiceUtils {

	/** Logger */
	private static Log logger = LogFactory.getLog(SindiceUtils.class);

	public static final int SINDICE_RESULTS_PER_PAGE = 10;

	/** base url for sindice search */
	// private final String baseURL =
	// "http://api.sindice.com/v2/search?qt=advanced&q=";
	public static final String SINDICE_URL = "http://sindice.com/api/v2/search?";

	/** sindice search link */
	public static final String RESULT_URL_PROPERTY = "http://sindice.com/vocab/search#link";

	public static final String FORMAT_PROPERTY = "http://sindice.com/vocab/fields#format";

	// urls that are banned for being too slow
	private static final Map<String, Long> bannedURLs = new HashMap<String, Long>();

	/**
	 * This method returns a String list of URLs that uniquely identify an RDF
	 * model. The data sources are uniquely identified by a String with a
	 * keyword as a String request.
	 * 
	 * @param query
	 *            valid sindice (v2) query (q).
	 * @param queryParams
	 *            sindice (v2) query parameters (qt (term,advanced,combined),
	 *            page, sortbydate (boolean), qv (if combined))
	 * @param user
	 *            the user making the request.
	 * @return a list of urls.
	 * @throws BlackbookSystemException
	 */
	public static Set<String> sindiceQuery(String query, int numResults,
			String[] supportedFormats) {

		// Timing info.
		long sindiceStartTime = System.currentTimeMillis();

		if (query == null || query.length() <= 0) {
			logger.warn("WARNING: No query specified");
			return new HashSet<String>();
		}

		Set<String> resultUrls = new HashSet<String>();

		try {

			for (int i = 1; resultUrls.size() < numResults; i++) {
				String sQuery = getURL(query, "qt=term&sortbydate=false"
						+ "&page=" + i);
				Model model = ModelFactory.createDefaultModel();
				model.read(sQuery);
				if (model.size() == 0) {
					break;
				}
//				StringWriter sw = new StringWriter();
//				model.write(sw);
//				logger.error(sw.toString());
				logger.info("number of statements in sindice response is: " + model.size());

				for (String format : supportedFormats) {
					if(resultUrls.size() >= numResults) {
						break;
					}
					ResIterator rItr = model.listResourcesWithProperty(model
							.createProperty(FORMAT_PROPERTY), model
							.createLiteral(format));
					while (rItr != null && rItr.hasNext() && resultUrls.size() < numResults) {
						Resource r = rItr.nextResource();
						StmtIterator it = model.listStatements(r, model
								.createProperty(RESULT_URL_PROPERTY),
								(Resource) null);
						while (it != null && it.hasNext() && resultUrls.size() < numResults) {
							Statement stmt = it.nextStatement();
							Resource link = stmt.getResource();
							if (link.isURIResource()) {
								logger.info("adding url from sindice query: "
										+ link.getURI());

								resultUrls.add(link.getURI());
							}
						}
					}
				}
			}

		} catch (Exception e) {
			logger.error("sindice query failed.", e);
		} finally {

		}

		long sindiceEndTime = System.currentTimeMillis();
		NumberFormat nf = NumberFormat.getInstance();
		logger
				.debug("Sindice keyword query for \""
						+ query
						+ "\n in "
						+ nf
								.format((double) (sindiceEndTime - sindiceStartTime) / 1000)
						+ " seconds.");

		logger.debug("Returning this many results: " + resultUrls.size());
		logger.debug("results are: " + resultUrls);
		return resultUrls;
	}

	/**
	 * Appends query and queryParams into a sindice (v2) search string and utf8
	 * encodes it.
	 * 
	 * @param query
	 *            The Sindice (v2) query (q)
	 * @param queryParams
	 *            The Sindice (v2) queryParams (qt (term,advanced,combined),
	 *            page, sortbydate (boolean), qv (if combined))
	 * @return
	 */
	public static String getURL(String query, String queryParams) {

		String urlQuery = "";

		try {
			query = "q=" + URLEncoder.encode(query, "utf-8");
		} catch (UnsupportedEncodingException e) {
			throw new RuntimeException(e);
		}

		if (StringUtils.isBlank(queryParams)) {
			urlQuery = query;
		} else {
			urlQuery = query + "&" + queryParams;
		}

		return SINDICE_URL + urlQuery;
	}

	public static boolean isTimelyRdfUrl(String url, final long timeout) {

		// if we have already failed at a longer timeout then don't try the
		// shorter time out... unless we are forced to try.
		if (bannedURLs.get(url) != null
				&& bannedURLs.get(url) > timeout) {
			return false;
		} else return true;

	}
	
	
	public static final Model readRdfUrl(String url, final long timeout, boolean force) {
		final String tempURL = url;
		final Model tempModel = ModelFactory.createDefaultModel();
		
		if (!force && bannedURLs.get(url) != null
				&& bannedURLs.get(url) > timeout) {
			return tempModel;
		}
		
		final AtomicBoolean isTimely = new AtomicBoolean(false);
		final AtomicBoolean stateSet = new AtomicBoolean(false);
		final Thread arpt = new Thread(new Runnable() {
			public void run() {
				// not great because we have to try to read it and then
				// really read it.
				logger.info("checking url for timeliness: " + tempURL);
				long startTime = System.currentTimeMillis();
				try {
					RDFReader jReader = tempModel.getReader("RDF/XML");
					InputStream is = new URL(tempURL).openStream();
					jReader.read(tempModel,is,tempURL);
					isTimely.set(true);
					stateSet.set(true);
					logger.info("url read in: "
							+ (System.currentTimeMillis() - startTime) / 1000 + " with num statements: " + tempModel.size());
				} 
				catch (Exception e) {

				}
			}
		});

		Thread killt = new Thread(new Runnable() {
			public void run() {
				try {
					int count = 0;
					while(count < timeout/1000) {
						Thread.sleep(1000);
						if(stateSet.get()) {
							break;
						}
						count++;
					}
				} catch (InterruptedException e) {
//					e.printStackTrace();
				}
				arpt.interrupt();
				stateSet.set(true);
			}
		});

		arpt.start();
		killt.start();
		
		long startTime = System.currentTimeMillis();
		while (!stateSet.get()) {
			try {
				Thread.currentThread().sleep(500);
//				logger.info("checking to see if arpt is still alive");
			} catch (InterruptedException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}
		logger.info("total read of url " +tempURL+ " took " + (System.currentTimeMillis() - startTime) / 1000 + " seconds.");

		arpt.interrupt();
		killt.interrupt();
		
		if (!isTimely.get()) {
			// check to see if it is worse than previously known
			logger.info(url + " was not successful read within the time limit.  Banning it for that time limit "+timeout+".  Force a reload by setting force to \"true\"");
			if (bannedURLs.get(url) == null || bannedURLs.get(url) != null && bannedURLs.get(url) < timeout) {
				bannedURLs.put(url, timeout);
			}
		} else {
			logger.info(url + " successfully read.");
			if(bannedURLs.get(url) != null && bannedURLs.get(url) > timeout) {
				bannedURLs.remove(url);
			}
		}

		return tempModel;
	}

	public static long getUrlBannedTimeThreshold(String url) {
		if(bannedURLs.get(url) == null) {
			return 0;
		}
		return bannedURLs.get(url);
	}
	
	public static final Map<String,Long> getBannedURLs() {
		return bannedURLs;
	}
}
